package com.jvm123

import groovy.sql.GroovyRowResult
import groovy.sql.Sql
import io.reactivex.rxjava3.core.ObservableEmitter
import io.reactivex.rxjava3.core.ObservableOnSubscribe

import java.util.function.Function
import java.util.stream.Collectors

/**
 * @author yawn
 */
class DataPublisher implements ObservableOnSubscribe<Collection<String>> {

    static final int STEP = 100

    Sql sql = Sql.newInstance(Constant.URL, Constant.USER, Constant.PWD, Constant.DRIVER)

    @Override
    void subscribe(ObservableEmitter<Collection<String>> emitter) throws Exception {
        Collection<Collection<String>> collection
        int offset = 0
        do {
            collection = getCollection(offset, STEP)
            for (Collection<String> strColl : collection) {
                emitter.onNext(strColl)
            }
            offset += STEP
        } while (collection.size() >= STEP)
        sql.close()
        emitter.onComplete()
    }

    Collection<Collection<String>> getCollection(int offset, int step) {
        List<GroovyRowResult> list = sql.rows("select * from order_goods where order_id in(select t.order_id from (select distinct order_id from order_goods limit ?, ?) t)", [offset, step])
        Map<String, List<GroovyRowResult>> orderMap = list.stream().collect(Collectors.groupingBy(new Function<GroovyRowResult, String>() {
            @Override
            String apply(GroovyRowResult row) {
                row.get("order_id")
            }
        }))
        Collection<Collection<String>> collection = new HashSet<>()
        for (entry in orderMap.entrySet()) {
            Collection<String> strColl = new HashSet<>()
            for (row in entry.getValue()) {
                String goodsId = row.get("goods_id")
                strColl.add(goodsId)
            }
            collection.add(strColl)
        }
        collection
    }
}